package com.example.bicycle.mqtt.impl;

import com.alibaba.fastjson.JSON;
import com.example.bicycle.mqtt.InitCallback;
import com.example.bicycle.mqtt.MQTTConnect;
import com.example.bicycle.mqtt.MqttMessageService;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Service("mqttMessage")
public class MqttMessageServiceImpl implements MqttMessageService {
    private final MQTTConnect mqttConnect;


    /**
     * 客户状态图
     */
    public static Map<String, Boolean> clientsStatusMap = new HashMap<>();
    /**
     * 离线消息缓存
     * key:topic
     * List<String>: message
     */
    Map<String, List<String>> offlineMessageCaching = new HashMap<>();

    @Autowired
    public MqttMessageServiceImpl(MQTTConnect mqttConnect) {
        this.mqttConnect = mqttConnect;

    }

    /**
     * 发布
     *
     * @param topic   主题
     * @param message 消息
     */
    public void publish(String topic, String message) {

        try {
            /* 判断是否是离线数据，是则缓存起来 */
            storeOfflineMessage(topic,message);
            mqttConnect.pub(topic, message);
        } catch (MqttException e) {
            System.out.println("发送失败");
            //失败缓存
            throw new RuntimeException(e);
        }
    }

    /**
     * 发布
     *
     * @param topic  主题
     * @param entity 实体
     */
    public <T> void publish(String topic, T entity) {
        String message = JSON.toJSONString(entity);
        try {
            /* 判断是否是离线数据，是则缓存起来 */
            storeOfflineMessage(topic,message);
            mqttConnect.pub(topic, message, 1);
        } catch (MqttException e) {
            System.out.println("发送失败");
            //失败缓存
            throw new RuntimeException(e);
        }
    }

    public void storeOfflineMessage(String topic, String message) {
        /* 先找到所有离线客户端，通过客户端ID去判断topic是否属于发给离线设备的，是则放入缓存集合中 */
        List<String> offlineClients = new ArrayList<>();
        for (Map.Entry<String, Boolean> entry : clientsStatusMap.entrySet()) {
            if (!entry.getValue()) {
                offlineClients.add(entry.getKey());
            }
        }
        /* 在clientStatusMap中找到value==false即掉线的*/
        for (String offlineClient : offlineClients) {
            /* 规定topic后面接clientId 如果结尾则缓存*/
            if (topic.endsWith(offlineClient)) {
                offlineMessageCaching.computeIfAbsent(topic, k -> new ArrayList<>());
                offlineMessageCaching.get(topic).add(message);
            }
        }
    }

    public void sendOfflineMessage(String clientId) {


        for (Map.Entry<String, List<String>> entry : offlineMessageCaching.entrySet()) {
            if (entry.getKey().endsWith(clientId)) {
                for (String message : entry.getValue()) {
                    try {
                        mqttConnect.pub(entry.getKey(), message);
                        Thread.sleep(50);
                    } catch (MqttException | InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }
                offlineMessageCaching.remove(entry.getKey());
            }
        }


    }
}
